package util;

/**
 * @author xiafan xiafan68@gmail.com
 */
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.Pipeline;

public class RedisCluster {

	Map<Integer, JedisPool> jedisPoolMap = new HashMap<Integer, JedisPool>();
	ThreadLocal<HashMap<Integer, Jedis>> jedisMap = new ThreadLocal<HashMap<Integer, Jedis>>();
	ThreadLocal<HashMap<Integer, Jedis>> pipeJedisMap = new ThreadLocal<HashMap<Integer, Jedis>>();
	ThreadLocal<HashMap<Integer, Pipeline>> pipelines = new ThreadLocal<HashMap<Integer, Pipeline>>();
	ThreadLocal<CountDownLatch> latch = new ThreadLocal<CountDownLatch>();
	private IPartition partition = null;

	private ExecutorService syncService = Executors.newCachedThreadPool();

	public RedisCluster(String[] hosts, int port) {
		JedisPoolConfig config = new JedisPoolConfig();
		config.setMaxActive(200);
		config.setMaxIdle(20);
		config.setMaxWait(1000l);
		for (int i = 0; i < hosts.length; i++) {
			jedisPoolMap.put(i, new JedisPool(config, hosts[i], port, 100 * 1000));
		}

		partition = PartitionFactory.createPartition(hosts);

	}

	public IPartition getPartition() {
		return partition;
	}

	public void close() {
		for (JedisPool pool : jedisPoolMap.values()) {
			pool.destroy();
		}
		jedisPoolMap.clear();
	}

	public Pipeline getPipeline(int nid) {
		Pipeline ret = null;
		if (getPipeJedis(nid) != null) {
			if (pipelines.get() == null) {
				pipelines.set(new HashMap<Integer, Pipeline>());
			}
			ret = pipelines.get().get(nid);
			if (ret == null) {
				ret = getPipeJedis(nid).pipelined();
				pipelines.get().put(nid, ret);
			}
		}
		return ret;
	}

	public void synAllPipelines() {
		if (pipelines.get() != null) {
			latch.set(new CountDownLatch(pipelines.get().size()));
			for (Pipeline pipeline : pipelines.get().values()) {
				syncService.submit(new SyncJob(pipeline, latch.get()));
			}
			while (true) {
				try {
					latch.get().await(); //一直等到所有的pipeline都执行完毕
					break;
				} catch (InterruptedException e) {
				}
			}
		}
	}

	public Jedis getJedis(int nid) {
		if (jedisMap.get() == null) {
			jedisMap.set(new HashMap<Integer, Jedis>());
		}

		Jedis ret = jedisMap.get().get(nid);
		if (ret == null && jedisPoolMap.containsKey(nid)) {
			ret = jedisPoolMap.get(nid).getResource();
			jedisMap.get().put(nid, ret);
		}
		return ret;
	}

	
	private Jedis getPipeJedis(int nid) {
		if (pipeJedisMap.get() == null) {
			pipeJedisMap.set(new HashMap<Integer, Jedis>());
		}

		Jedis ret = pipeJedisMap.get().get(nid);
		if (ret == null && jedisPoolMap.containsKey(nid)) {
			ret = jedisPoolMap.get(nid).getResource();
			pipeJedisMap.get().put(nid, ret);
		}
		return ret;
	}
	
	public void putJedis(int nid, Jedis jedis) {
		if (jedisPoolMap.containsKey(nid)) {
			jedisPoolMap.get(nid).returnResource(jedis);
		}
	}

	public void returnAllJedis() {
		if (jedisMap.get() != null) {
			for (Entry<Integer, Jedis> entry : jedisMap.get().entrySet()) {
				putJedis(entry.getKey(), entry.getValue());
			}
			jedisMap.get().clear();
		}
		
		if (pipeJedisMap.get() != null) {
			for (Entry<Integer, Jedis> entry : pipeJedisMap.get().entrySet()) {
				putJedis(entry.getKey(), entry.getValue());
			}
			pipeJedisMap.get().clear();
		}
	}

	private static class SyncJob implements Runnable {
		private Pipeline pipe;
		private CountDownLatch latch;

		public SyncJob(Pipeline pipe_, CountDownLatch latch_) {
			this.pipe = pipe_;
			this.latch = latch_;
		}

		@Override
		public void run() {
			pipe.sync();
			latch.countDown();
		}
	}
}
